/****************************************************************************
 *
 * Copyright (C) 2003-2008 Los Alamos National Security, LLC
 *                         Packet Analytics Corporation
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License Version 2 as
 * published by the Free Software Foundation.  You may not use, modify or
 * distribute this program under any other version of the GNU General
 * Public License.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 *
 ****************************************************************************/
package nfse.query;

import nfse.*;

import java.io.*;
import java.util.*;
import java.sql.*; // import com.sleepycat.je.*;
// import com.sleepycat.db.*;
// import com.sleepycat.je.*;

import java.util.zip.*;

/**
 * @author bduphoff
 * 
 *         TODO To change the template for this generated type comment go to
 *         Window - Preferences - Java - Code Style - Code Templates
 */
public class SearchEngine {

	private MetaDatabase mdb = null;

	private Vector<Integer> fileIDs = null;

	private Vector<String> filePaths = null;

	private String offsetTable = null;

	private RandomAccessFile dataIn = null;

	private GZIPInputStream dataInGZipped = null;

	private long fileOffset = 0;

	private String dataFilePath = "";

	private String[] joinTables = null;

	// private int type = -1;

	String zipOnExit = "";

	NetFSERecord rec = null;

	Connection tempConnection = null;

	ResultSet rs = null;

	int limit = 50000;

	int offset = 0;

	public SearchEngine(MetaDatabase mdb, Connection tempConnection,
			NetFSERecord rec) {
		this.mdb = mdb;
		this.tempConnection = tempConnection;
		this.rec = rec;
	}

	public void init() {
		try {

			if (dataIn != null) {
				dataIn.close();
			}
			if (dataInGZipped != null)
				dataInGZipped.close();
			if (zipOnExit.length() > 0) {
				System.out.println("Compressing: " + zipOnExit);
				NetFSEUtil.execAndWait("gzip " + zipOnExit);
			}

			offset = 0;

			Statement tempStatement = tempConnection.createStatement();
			if (offsetTable != null) {
				try {
					String sql = "drop table " + offsetTable;
					tempStatement.execute(sql);
				} catch (Exception e) {
					// no op
				}
			}
			if (joinTables != null) {
				for (int i = 0; i < joinTables.length; i++) {
					if (joinTables[i] != null) {
						try {
							String sql = "drop table " + joinTables[i];
							tempStatement.execute(sql);
						} catch (Exception E) {
							// no op
						}
					}
				}
			}
			tempStatement.close();

		} catch (Exception E) {
			E.printStackTrace();
		}
	}

	public void close() {
		init();
		try {
			tempConnection.close();
		} catch (Exception e) {

		}
	}

	public int performSearch(int month, int day, int year, int run, int type,
			int sensor, GlobalField[] fields, String[] values,
			String[] values2, int[] predicates) {
		try {
			if (fields.length == 1) {
				return performSearch(month, day, year, run, type, sensor,
						fields[0], values[0], values2[0], predicates[0], false);
			}

			int numFields = fields.length;
			if (values2[values2.length - 1] == null)
				numFields--;

			joinTables = new String[numFields];

			//boolean emptyQuery = false;
			Statement tempStatement = tempConnection.createStatement();
			Statement statement = mdb.createStatement();
			for (int i = 0; i < numFields; i++) {

				String sql = "select DB_Path from Static_Indices where  Field_ID = "
						+ fields[i].getId()
						+ " and Sensor_ID = "
						+ sensor
						+ " and Index_Date='"
						+ year
						+ "-"
						+ month
						+ "-"
						+ day
						+ "' and Run=" + run;
				System.out.println("SQL: " + sql);

				ResultSet rs = statement.executeQuery(sql);
				if (rs.next()) {

					// The data is indexed, search
					sql = "select nextval('searchid')";
					ResultSet rs1 = tempStatement.executeQuery(sql);
					rs1.next();
					joinTables[i] = "static" + rs1.getInt(1);
					rs1.close();

					tempStatement.execute("create table " + joinTables[i]
							+ " (offset_hex varchar(8))");

					String indexPath = rs.getString("DB_Path");
					rs.close();
					IndexInfo info = IndexInfo.processInfo(indexPath);
					if (info != null) {
						this.dataFilePath = info.sessionFile;
						searchIndex(info, values[i], values2[i],
								predicates[i], joinTables[i], tempStatement,
								true);
					}
				} else {
					rs.close();
					// Not indexed, search raw data, for now noop
					System.out.println("Not indexed: " + sql);
				}
			}

			statement.close();
			tempStatement.close();
		} catch (Exception E) {
			E.printStackTrace();
		}
		return -1;
	}

	private int performSearch(int month, int day, int year, int run, int type,
			int sensor, GlobalField field, String value, String value2,
			int predicate, boolean isJoin) {
		try {
			Statement statement = mdb.createStatement();

			Statement tempStatement = tempConnection.createStatement();

			String sql = "select DB_Path from Static_Indices where  Field_ID = "
					+ field.getId()
					+ " and Sensor_ID = "
					+ sensor
					+ " and Index_Date='"
					+ year
					+ "-"
					+ month
					+ "-"
					+ day
					+ "' and Run=" + run;
			rs = statement.executeQuery(sql);
			if (rs.next()) {
				// The data is indexed, search
				String indexPath = rs.getString("DB_Path");
				rs.close();
				IndexInfo info = IndexInfo.processInfo(indexPath);
				this.dataFilePath = info.sessionFile;
				searchIndex(info, value, value2, predicate,
						offsetTable, tempStatement, false);

			} else {
				// Not indexed, search raw data, for now noop

			}
			rs.close();
			statement.close();
			return 0;
		} catch (Exception E) {
			E.printStackTrace();
		}
		return -1;
	}

	private long searchIndex(IndexInfo info, String value, String value2,
			int predicate, String offsetTable, Statement tempStatement,
			boolean isJoin) {
		try {
			if ((predicate == NetFSE.PREDICATE_RANGE) && (value2.length() == 0)) {
				System.err
						.println("Range predicate specified with null upper bound.");
				return -2;
			}

			fileIDs = new Vector<Integer>();
			filePaths = new Vector<String>();
			String fileIndexPath = info.indexPath + "/file.index";
			LineNumberReader in = new LineNumberReader(new FileReader(
					fileIndexPath));
			String line = in.readLine();
			while (line != null) {
				if (line.length() > 4) {
					fileIDs.addElement(new Integer(Integer.parseInt(line
							.substring(0, 4), 16)));
					filePaths.addElement(line.substring(4));
					// System.out.println(line);
				}
				line = in.readLine();
			}
			in.close();

			System.out.println("searchIndex: " + value + "," + value2);

			String path = info.indexPath;
			long numResults = 0;
			long equalOffset = getOffset(path + "/index/root.page", value,
					info.keyLength);
			// System.out.println("Equal offset=" + equalOffset);
			long prevOffset = -1;
			long nextOffset = -1;
			switch (predicate) {
			case NetFSE.PREDICATE_EQUAL_TO:
				if (equalOffset == -1) {
					// System.out.println("No matching records: " +
					// info.toString() + ", " + value );
					return 0;
				}
				numResults = equalTo(info, equalOffset, offsetTable,
						tempStatement);
				break;
			case NetFSE.PREDICATE_LESS_THAN:
				if (equalOffset == -1) {
					prevOffset = getPrevOffset(path + "/index/root.page",
							value, info.keyLength);
					if (prevOffset == -1) {
						// System.out.println("No matching records.");
						return 0;
					}
					numResults = equalTo(info, prevOffset, offsetTable,
							tempStatement);

					numResults += lessThan(info, prevOffset, value,
							offsetTable, tempStatement);
				} else {
					numResults = lessThan(info, equalOffset, value,
							offsetTable, tempStatement);
				}
				break;
			case NetFSE.PREDICATE_GREATER_THAN:
				if (equalOffset == -1) {
					nextOffset = getNextOffset(path + "/index/root.page",
							value, info.keyLength);
					if (nextOffset == -1) {
						// System.out.println("No matching records.");
						return 0;
					}
					numResults = equalTo(info, nextOffset, offsetTable,
							tempStatement);

					numResults += greaterThan(info, nextOffset, offsetTable,
							tempStatement);
				} else {
					numResults = greaterThan(info, equalOffset, offsetTable,
							tempStatement);
				}
				break;
			case NetFSE.PREDICATE_LESS_THAN_EQUAL_TO:
				if (equalOffset == -1) {
					prevOffset = getPrevOffset(path + "/index/root.page",
							value, info.keyLength);
					if (prevOffset == -1) {
						// System.out.println("No matching records.");
						return 0;
					}
					numResults = lessThan(info, prevOffset, value, offsetTable,
							tempStatement);

					numResults += equalTo(info, prevOffset, offsetTable,
							tempStatement);
				} else {
					numResults = equalTo(info, equalOffset, offsetTable,
							tempStatement);

					numResults += lessThan(info, equalOffset, value,
							offsetTable, tempStatement);
				}
				break;
			case NetFSE.PREDICATE_GREATER_THAN_EQUAL_TO:
				if (equalOffset == -1) {
					nextOffset = getNextOffset(path + "/index/root.page",
							value, info.keyLength);
					if (nextOffset == -1) {
						// System.out.println("No matching records.");
						return 0;
					}
					numResults = greaterThan(info, nextOffset, offsetTable,
							tempStatement);

					numResults += equalTo(info, nextOffset, offsetTable,
							tempStatement);
				} else {
					numResults = equalTo(info, equalOffset, offsetTable,
							tempStatement);

					numResults += greaterThan(info, equalOffset, offsetTable,
							tempStatement);
				}
				break;
			case NetFSE.PREDICATE_RANGE:
				long offset2 = getOffset(path + "/index/root.page", value2,
						info.keyLength);
				if (offset2 == -1) {
					offset2 = getPrevOffset(path + "/index/root.page", value2,
							info.keyLength);
				}
				if (equalOffset == -1) {
					System.out.println("No equal offset, getting next");
					equalOffset = getNextOffset(path + "/index/root.page",
							value, info.keyLength);
				} else {

				}
				System.out.println(equalOffset + ", " + offset2 + ", "
						+ (equalOffset > offset2));
				if ((equalOffset == -1) || (offset2 == -1)
						|| (equalOffset > offset2)) {
					System.out.println("No matching records.");
					return 0;
				}
				numResults = range(info, equalOffset, offset2, tempStatement,
						offsetTable, value, value2);
				break;
			} // End predicate test switch
			return numResults;
		} catch (Exception E) {
			E.printStackTrace();
		}
		return -1;
	}

	private int outputOffset(byte[] offset, String table,
			Statement tempStatement) {
		try {

			String sql = "insert into " + table + " values ('"
					+ new String(offset) + "')";
			tempStatement.execute(sql);

			// emptyDbt = new DatabaseEntry();
			// emptyDbt.setData(" ".getBytes());
			// DatabaseEntry offsetDbt = new DatabaseEntry(offset);
			// System.out.println(new String(offset) + ", " + Long.parseLong(new
			// String(offset),16));
			// bdb.put(null, offsetDbt, emptyDbt);
		} catch (Exception E) {
			E.printStackTrace();
			return -1;
		}
		return 0;
	}

	private long range(IndexInfo info, long offset, long offset2,
			Statement tempStatement, String table, String value1, String value2) {
		long numResults = 0;
		RandomAccessFile in = null;
		try {
			// System.out.println("RANGE: " + offset + ", " + offset2);
			String indexDataPath = info.indexPath + "/index.data";
			in = new RandomAccessFile(indexDataPath, "r");
			in.seek(offset);
			boolean done = false;
			boolean done2 = false;
			byte ch = (byte) ' ';
			int count = 0;
			int count2 = 0;
			String temp = "";
			String temp2 = "";
			long tempOffset = 0;
			while (!done) {
				temp = "";
				count = 0;
				tempOffset = in.getFilePointer();
				if (info.keyLength != -1) { // Fixed length field
					do {
						ch = in.readByte();
						count++;
						temp += (char) ch;
					} while ((count < info.keyLength) && (ch != '\n'));
					// System.out.println("temp: '" + temp + "', value2: '"
					// + value2 + "'");
					if ((temp.compareTo(value1) < 0)
							|| (temp.compareTo(value2) > 0)) {
						in.close();
						return numResults;
					}

				} else { // Variable length field, currently not supported
					in.close();
					return 0;
				}
				if ((temp.length() == info.keyLength)
						&& (tempOffset <= offset2)) {
					count2 = 0;
					temp2 = "";
					done2 = false;
					while (!done2) {
						do {
							ch = in.readByte();
							temp2 += (char) ch;
							count2++;
						} while ((count2 < 8) && (ch != '\n'));
						if ((ch == '\n') || (count2 < 8)) {
							done2 = true;
						}
						if (count2 == 8) {
							outputOffset(temp2.getBytes(), table, tempStatement);
							numResults++;
						}
						count2 = 0;
						temp2 = "";
					}
				} else {
					done = true;
				}
			}
			in.close();
		} catch (Exception E) {
			try {
				in.close();
			} catch (Exception E2) {
				E2.printStackTrace();
			}
		}
		return numResults;
	}

	private long greaterThan(IndexInfo info, long offset, String table,
			Statement tempStatement) {
		RandomAccessFile in = null;
		long numResults = 0;
		try {
			String indexDataPath = info.indexPath + "/index.data";
			in = new RandomAccessFile(indexDataPath, "r");
			in.seek(offset);
			byte ch = in.readByte();
			while (ch != '\n')
				ch = in.readByte();
			boolean done = false;
			while (!done) {
				String temp = "";
				int count = 0;
				if (info.keyLength != -1) { // Fixed length field
					do {
						try {
							ch = in.readByte();
						} catch (IOException ioe) {
							// ioe.printStackTrace();
							ch = '\n';
						}
						count++;
						temp += (char) ch;
					} while ((count < info.keyLength) && (ch != '\n'));
				} else { // Variable length field
					in.close();
					return 0;
				}
				if (temp.length() == info.keyLength) {
					int count2 = 0;
					String temp2 = "";
					boolean done2 = false;
					while (!done2) {
						do {
							try {
								ch = in.readByte();
							} catch (IOException ioe) {
								// ioe.printStackTrace();
								ch = '\n';
							}
							temp2 += (char) ch;
							count2++;
						} while ((count2 < 8) && (ch != '\n'));
						if ((ch == '\n') || (count2 < 8)) {
							done2 = true;
						}
						if (count2 == 8) {
							outputOffset(temp2.getBytes(), offsetTable,
									tempStatement);
							numResults++;

						}
						count2 = 0;
						temp2 = "";
					}
				} else {
					done = true;
				}
			}
			in.close();
		} catch (Exception E) {
			try {
				in.close();
			} catch (Exception E2) {
				E2.printStackTrace();
			}
		}
		return numResults;
	}

	private long lessThan(IndexInfo info, long offset, String value,
			String table, Statement tempStatement) {
		long numResults = 0;
		RandomAccessFile in = null;
		try {
			if (offset == 0)
				return 0;
			String indexDataPath = info.indexPath + "/index.data";
			in = new RandomAccessFile(indexDataPath, "r");
			in.seek(0);
			boolean done = false;
			byte ch = 0;
			while (!done) {
				String temp = "";
				int count = 0;
				if (info.keyLength != -1) { // Fixed length field
					do {
						ch = in.readByte();
						count++;
						temp += (char) ch;
					} while ((count < info.keyLength) && (ch != '\n'));
				} else { // Variable length field
					in.close();
					return 0;
				}
				if ((temp.length() == info.keyLength)
						&& (temp.compareTo(value) < 0)) {
					int count2 = 0;
					String temp2 = "";
					boolean done2 = false;
					while (!done2) {
						do {
							ch = in.readByte();
							temp2 += (char) ch;
							count2++;
						} while ((count2 < 8) && (ch != '\n'));
						if ((ch == '\n') || (count2 < 8)) {
							done2 = true;

						}
						if (count2 == 8) {
							outputOffset(temp2.getBytes(), table, tempStatement);
							numResults++;

						}
						count2 = 0;
						temp2 = "";
					}
				} else {
					done = true;
				}
			}
			in.close();
		} catch (Exception E) {
			try {
				in.close();
			} catch (Exception E2) {
				E2.printStackTrace();
			}
		}
		return numResults;
	}

	private long equalTo(IndexInfo info, long offset, String table,
			Statement tempStatement) {
		long numResults = 0;
		RandomAccessFile in = null;
		try {
			String indexDataPath = info.indexPath + "/index.data";
			in = new RandomAccessFile(indexDataPath, "r");
			byte ch = 0;
			if (info.keyLength != -1) { // Fixed length field
				in.seek(offset + info.keyLength);
				ch = in.readByte();
			} else { // Variable length field
				in.close();
				return 0;
			}
			String temp = "";
			int count = 0;
			do {
				count++;
				temp += (char) ch;
				if (count == 8) {
					outputOffset(temp.getBytes(), table, tempStatement);
					numResults++;
					count = 0;
					temp = "";
				}
				ch = in.readByte();
			} while (ch != '\n');
			in.close();
		} catch (Exception E) {
			try {
				in.close();
			} catch (Exception E2) {
				E2.printStackTrace();
			}
		}
		return numResults;
	}

	private long getPrevOffset(String pagePath, String key, int keyLength) {
		long offset = -1;
		try {
			// System.out.println("Checking: " + pagePath);
			LineNumberReader in = new LineNumberReader(new FileReader(pagePath));
			int leafValue = pagePath.indexOf("/page");
			boolean isLeaf = false;
			if (leafValue >= 0) {
				isLeaf = true;
			}
			String lastKey = "";
			String lastValue = "";
			String line = in.readLine();
			while (line != null) {
				if (line.length() < keyLength) {
					if (isLeaf) {
						return Long.parseLong(lastValue, 16);
					} else {
						in.close();
						String nextPage = getFilePath(Integer.parseInt(
								lastValue, 16));
						return getPrevOffset(nextPage, key, keyLength);
					}
				}
				String tempKey = line.substring(0, keyLength);
				String tempValue = line.substring(keyLength);
				// System.out.println(tempKey + "," + tempValue);
				if (lastKey.length() == 0) {
					lastKey = tempKey;
					lastValue = tempValue;
				}
				if ((isLeaf) && (tempKey.compareTo(key) >= 0)) { // Return
					// the
					// last
					// offset found
					in.close();
					return Long.parseLong(lastValue, 16);
				} else if ((!isLeaf) && (key.equals(tempKey))) {
					in.close();
					String nextPage = getFilePath(Integer.parseInt(tempValue,
							16));
					return getPrevOffset(nextPage, key, keyLength);
				} else if ((!isLeaf)
						&& ((key.compareTo(lastKey) > 0) && (key
								.compareTo(tempKey) < 0))) {
					in.close();
					String nextPage = getFilePath(Integer.parseInt(lastValue,
							16));
					return getPrevOffset(nextPage, key, keyLength);
				}
				lastKey = tempKey;
				lastValue = tempValue;
				line = in.readLine();
			}
			in.close();
			if (isLeaf) {
				return Long.parseLong(lastValue, 16);
			} else {
				in.close();
				String nextPage = getFilePath(Integer.parseInt(lastValue, 16));
				return getPrevOffset(nextPage, key, keyLength);
			}
		} catch (Exception E) {
			E.printStackTrace();

		}
		return offset;
	}

	private long getNextOffset(String pagePath, String key, int keyLength) {
		long offset = -1;
		try {
			LineNumberReader in = new LineNumberReader(new FileReader(pagePath));
			int leafValue = pagePath.indexOf("/page");
			boolean isLeaf = false;
			if (leafValue >= 0) {
				isLeaf = true;
			}
			String lastValue = "0001";
			String lastKey = "";
			String line = in.readLine();
			// System.out.println(key + ", " + keyLength);
			// System.out.println(pagePath);

			while (line != null) {
				System.out.println(line);
				if (line.length() < keyLength) {
					if (isLeaf) {
						in.close();
						return Long.parseLong(lastValue, 16);
					} else {
						in.close();
						String nextPage = getFilePath(Integer.parseInt(
								lastValue, 16));
						return getNextOffset(nextPage, key, keyLength);
					}
				}
				String tempKey = line.substring(0, keyLength);
				String tempValue = line.substring(keyLength);
				if (isLeaf) {
					if (tempKey.equals(key)) {
						line = in.readLine();
						if (line == null) {
							in.close();
							return -1;
						}
						if (line.length() == 8 + keyLength) {
							String offsetStr = line.substring(keyLength, line
									.length()
									- keyLength);
							in.close();
							return Long.parseLong(offsetStr, 16);
						} else {
							in.close();
							return -1;
						}
					} else if (tempKey.compareTo(key) > 0) {
						in.close();
						return Long.parseLong(tempValue, 16);
					}
				} else { // not a leaf page
					if (tempKey.compareTo(key) >= 0) {
						in.close();
						String nextPage = getFilePath(Integer.parseInt(
								lastValue, 16));
						return getNextOffset(nextPage, key, keyLength);
					} else if ((tempKey.compareTo(key) < 0)
							&& (key.compareTo(lastKey) > 0)) {
						in.close();
						String nextPage = getFilePath(Integer.parseInt(
								tempValue, 16));
						return getNextOffset(nextPage, key, keyLength);
					}

				}
				lastValue = tempValue;
				lastKey = tempKey;
				line = in.readLine();
			}
			in.close();

			String nextPage = getFilePath(Integer.parseInt(lastValue, 16));
			return getOffset(nextPage, key, keyLength);

		} catch (Exception E) {
			E.printStackTrace();
		}
		return offset;
	}

	private long getOffset(String pagePath, String key, int keyLength) {
		long offset = -1;
		try {
			System.out.println("Current page: " + pagePath);

			LineNumberReader in = new LineNumberReader(new FileReader(pagePath));
			int leafValue = pagePath.indexOf("/page");
			boolean isLeaf = false;
			if (leafValue >= 0) {
				isLeaf = true;
			}

			String lastKey = "";
			String lastValue = "0001";
			//int delimLoc = -1;

			String line = in.readLine();
			while (line != null) {
				// System.out.println("page entry: '" + line + "'");
				if ((line.length() > keyLength) && (keyLength > 0)) {
					String tempKey = "";
					String tempValue = "";
					tempKey = line.substring(0, keyLength);
					tempValue = line.substring(keyLength);
					// System.out.println("Key: " + tempKey + ", Value: " +
					// tempValue);
					if (lastKey.length() == 0) {
						lastKey = tempKey;
						lastValue = tempValue;
					}
					if (isLeaf && (key.equals(tempKey))) { // Return the first
						// matching offset
						in.close();
						return Long.parseLong(tempValue, 16);
					} else if ((isLeaf) && (tempKey.compareTo(key) > 0)) { // No
						// matching
						// values
						in.close();
						return -1;
					} else if ((!isLeaf) && (key.equals(tempKey))) {
						in.close();
						String nextPage = getFilePath(Integer.parseInt(
								tempValue, 16));
						return getOffset(nextPage, key, keyLength);
					} else if ((!isLeaf) && (key.compareTo(lastKey) > 0)
							&& (key.compareTo(tempKey) < 0)) {
						in.close();
						String nextPage = getFilePath(Integer.parseInt(
								lastValue, 16));
						return getOffset(nextPage, key, keyLength);
					}
					lastKey = tempKey;
					lastValue = tempValue;
				} else {
					if (!isLeaf) {
						in.close();
						String nextPage = getFilePath(Integer.parseInt(
								lastValue, 16));
						// System.out.println("Next page: " + nextPage);
						return getOffset(nextPage, key, keyLength);
					}
				}
				line = in.readLine();
			}
			in.close();

			String nextPage = getFilePath(Integer.parseInt(lastValue, 16));
			return getOffset(nextPage, key, keyLength);

		} catch (Exception E) {
			E.printStackTrace();
		}
		return offset;
	}

	private String getFilePath(int fileID) {
		for (int i = 0; i < fileIDs.size(); i++)
			if (((Integer) fileIDs.elementAt(i)).intValue() == fileID) {
				return (String) filePaths.elementAt(i);
			}
		return null;
	}

	private String getRecordAtOffset(long offset) {
		try {
			if (dataIn != null) {
				dataIn.seek(offset);
				// System.out.println("OFFSET=" + offset);
				return rec.getRecordLine(dataIn);
			} else if (dataInGZipped != null) {

				long toSkip = offset - fileOffset;
				long skipped = 0;
				while (skipped != toSkip) {
					skipped += dataInGZipped.skip(toSkip);
					fileOffset += skipped;
					// System.out.println("OFFSET=" + offset +", FILEOFFSET=" +
					// fileOffset);
				}
				// System.out.println("FOUND IT ! OFFSET=" + offset
				// +", FILEOFFSET=" + fileOffset);
				String line = rec.getRecordLine(dataInGZipped);
				fileOffset += line.getBytes().length;
				return line;
			}
		} catch (Exception E) {
			E.printStackTrace();
		}
		return null;
	}

	public String getNextRecord(Statement tempStatement) {
		try {
			if ((dataIn == null) && (dataInGZipped == null))
				openDataFile(dataFilePath);

			if (rs == null) {
				String sql = "select t1.offset_hex from ";
				for (int i = 0; i < joinTables.length; i++) {
					if (i > 0)
						sql += ",";
					sql += joinTables[i] + " t" + (i + 1);
				}
				if (joinTables.length > 1) {
					sql += " where ";
					for (int i = 1; i < joinTables.length; i++) {
						if (i > 1)
							sql += " and ";
						sql += " t1.offset_hex = " + " t" + (i + 1)
								+ ".offset_hex ";
					}
				}
				sql += " order by t1.offset_hex asc limit " + limit
						+ " offset " + offset;
				rs = tempStatement.executeQuery(sql);

			}

			if (rs.next()) {

				// return null;
			} else {
				offset += limit;
				rs.close();
				String sql = "select t1.offset_hex from ";
				for (int i = 0; i < joinTables.length; i++) {
					if (i > 0)
						sql += ",";
					sql += joinTables[i] + " t" + (i + 1);
				}
				if (joinTables.length > 1) {
					sql += " where ";
					for (int i = 1; i < joinTables.length; i++) {
						if (i > 1)
							sql += " and ";
						sql += " t1.offset_hex = " + " t" + (i + 1)
								+ ".offset_hex ";
					}
				}
				sql += " order by t1.offset_hex asc limit " + limit
						+ " offset " + offset;
				rs = tempStatement.executeQuery(sql);
				if (!rs.next()) {
					rs.close();
					return null;
				}
			}

			String offStr = rs.getString(1);
			String rec = getRecordAtOffset(Long.parseLong(offStr, 16));
			// System.out.println(rec);
			return rec;

		} catch (Exception E) {
			E.printStackTrace();

		}
		return null;
	}

	private void openDataFile(String dataPath) throws Exception {
		zipOnExit = "";
		File dataFile = new File(dataPath);
		File dataFileGZipped = new File(dataPath + ".gz");
		while ((dataFile.exists()) && (dataFileGZipped.exists())) {
			Thread.sleep(1000);
		}
		if (dataFile.exists() && dataFile.getName().endsWith(".gz")) {
			dataInGZipped = new GZIPInputStream(new FileInputStream(dataFile));
			fileOffset = 0;
		} else if (dataFile.exists()) {
			dataIn = new RandomAccessFile(dataPath, "r");
		} else if (dataFileGZipped.exists()) {
			dataInGZipped = new GZIPInputStream(new FileInputStream(
					dataFileGZipped));
			fileOffset = 0;

			// }

		}
	}
}